package com.smart.tool.scheduler.impl;

import com.cronutils.model.definition.CronDefinition;
import com.cronutils.model.definition.CronDefinitionBuilder;
import com.cronutils.model.time.ExecutionTime;
import com.cronutils.parser.CronParser;
import com.smart.tool.scheduler.*;
import io.netty.util.HashedWheelTimer;
import io.netty.util.Timeout;
import io.netty.util.TimerTask;
import org.apache.commons.lang3.StringUtils;

import java.time.Duration;
import java.time.ZonedDateTime;
import java.util.Date;
import java.util.Optional;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @Author xiaoqiang
 * @Version SmartScheduler.java, v 0.1 2025年08月13日 19:26 xiaoqiang
 * @Description: * 基于 Netty HashedWheelTimer 的增强型通用定时任务调度器。
 * * 功能：
 * * - 支持一次性、固定间隔、斐波那契间隔任务。
 * * - 支持最大执行次数、任务取消、任务关闭。
 * * - 任务执行结束或取消后自动从注册表移除，保障内存安全。
 * * - 新增：限制最大待处理任务数量。
 * * - 新增：防止具有相同键的任务重复添加。
 */
public class SmartSchedulerImpl implements SmartScheduler {

    private static final Logger log = LoggerFactory.getLogger(SmartSchedulerImpl.class);


    // --- 常量定义 ---
    // 定义标准 Unix Cron 表达式格式 (秒 分 时 日 月 周)
    private static final CronDefinition CRON_DEFINITION = CronDefinitionBuilder.defineCron()
            .withSeconds().and()
            .withMinutes().and()
            .withHours().and()
            .withDayOfMonth().supportsL().supportsW().supportsLW().supportsQuestionMark().and()
            .withMonth().and()
            .withDayOfWeek().withValidRange(1, 7).withMondayDoWValue(2).supportsHash().supportsL().supportsQuestionMark().and()
            .withYear().optional().and()
            .instance();

    private static final CronParser CRON_PARSER = new CronParser(CRON_DEFINITION);

    // --- 实例字段 ---
    private final HashedWheelTimer hashedWheelTimer; // 移除 static
    private final int maxPendingTasks;
    private final ConcurrentHashMap<String, TaskWrapper> taskRegistry = new ConcurrentHashMap<>(); // 移除 static


    /**
     * 构造函数
     *
     * @param tickDuration  时间轮每格的时长
     * @param unit          时间单位
     * @param ticksPerWheel 时间轮的格数
     */
    public SmartSchedulerImpl(long tickDuration, TimeUnit unit, int ticksPerWheel) {
        this(tickDuration, unit, ticksPerWheel, 0);
    }

    public SmartSchedulerImpl(long tickDuration, TimeUnit unit, int ticksPerWheel, int maxPendingTasks, ThreadFactory threadFactory) {
        // 移除 static
        this.hashedWheelTimer = new HashedWheelTimer(threadFactory, tickDuration, unit, ticksPerWheel);
        this.maxPendingTasks = maxPendingTasks;
    }

    /**
     * 构造函数
     *
     * @param tickDuration    时间轮每格的时长
     * @param unit            时间单位
     * @param ticksPerWheel   时间轮的格数
     * @param maxPendingTasks 最大待处理任务数，<=0 表示无限制
     */
    public SmartSchedulerImpl(long tickDuration, TimeUnit unit, int ticksPerWheel, int maxPendingTasks) {
        ThreadFactory threadFactory = r -> {
            Thread t = new Thread(r, "SmartScheduler-Timer");
            t.setDaemon(true);
            return t;
        };

        // 移除 static
        this.hashedWheelTimer = new HashedWheelTimer(threadFactory, tickDuration, unit, ticksPerWheel);
        this.maxPendingTasks = maxPendingTasks;
    }

    // --- 任务包装器 (核心逻辑) ---
    private class TaskWrapper<T> implements TimerTask { // 移除 static
        final String taskId;
        final String taskKey; // 用于防重的键
        final Task<T> task;
        final TaskType type;
        final long intervalNanos; // 固定间隔或斐波那契初始间隔
        final String cronExpressionStr; // Cron 表达式字符串
        final ExecutionTime cronExecutionTime; // 解析后的 Cron 执行时间计算器
        final int maxExecutions;
        final AtomicInteger executionCount = new AtomicInteger(0);
        final AtomicLong statusOrdinal = new AtomicLong(TaskStatus.SCHEDULED.ordinal());
        long fibPrevNanos = 0;
        long fibCurrNanos = TimeUnit.MILLISECONDS.toNanos(1); // 初始为1ms
        volatile Timeout timeout;
        TaskCallback taskCallback;

        final CompletableFuture<T> resultFuture = new CompletableFuture<>();

        TaskWrapper(String taskId, String taskKey, Task<T> task, TaskType type,
                    long intervalNanos, String cronExpressionStr, int maxExecutions, TaskCallback taskCallback) {
            this.taskId = taskId;
            this.taskKey = taskKey;
            this.task = task;
            this.type = type;
            this.intervalNanos = intervalNanos;
            this.cronExpressionStr = cronExpressionStr;
            this.maxExecutions = maxExecutions;
            this.cronExecutionTime = (type == TaskType.CRON && cronExpressionStr != null) ?
                    ExecutionTime.forCron(CRON_PARSER.parse(cronExpressionStr)) : null;
            this.taskCallback = taskCallback;
        }

        private void processSuspended() {


        }

        @Override
        public void run(Timeout timeout) throws Exception {
            long oldStatusOrdinal = statusOrdinal.get();
            TaskStatus oldStatus = TaskStatus.fromOrdinal(oldStatusOrdinal);
            if (oldStatus != TaskStatus.SCHEDULED) {
                cleanup(taskId);
                return;
            }
            if (!statusOrdinal.compareAndSet(TaskStatus.SCHEDULED.ordinal(), TaskStatus.RUNNING.ordinal())) {
                return;
            }

            ScheduleContext<T> context = ScheduleContext.of(taskId, taskKey, type);
            context.withStartTime(new Date());
            boolean shouldReschedule = false;
            try {
                if (task instanceof RunnableTask) {
                    ((RunnableTask) task).execute(context);
                }
                if (task instanceof CallableTask) {
                    T result = ((CallableTask<T>) task).execute(context);
                    context.withResult(result);
                }
            } catch (Throwable e) {
                context.withThrowable(e);
                if (taskCallback != null) {
                    taskCallback.onException(context, e);
                }
                log.error("Error in task " + taskId + " (key: " + taskKey + "): " + e.getMessage(), e);
            } finally {
                context.withExecutionCount(getExecutionCount() + 1);
                int count = executionCount.incrementAndGet();
                //如果超过最大的执行次数，则是人为任务结束
                boolean isCompleted = (maxExecutions > 0 && count >= maxExecutions);
                oldStatusOrdinal = statusOrdinal.get();
                oldStatus = TaskStatus.fromOrdinal(oldStatusOrdinal);

                shouldReschedule = changeStatusAfterExecute(isCompleted, oldStatus, shouldReschedule);
                if (shouldReschedule) {
                    long delayNanos = calculateNextDelayNanos();
                    if (delayNanos >= 0) {
                        Date now = new Date();
                        Date nextExecTime = new Date(now.getTime() + TimeUnit.NANOSECONDS.toMillis(delayNanos));
                        context.withNextExecutionTime(nextExecTime);

                        TaskStatus currentStatus = TaskStatus.fromOrdinal(statusOrdinal.get());
                        if (currentStatus == TaskStatus.SCHEDULED) {
                            this.timeout = timeout.timer().newTimeout(this, delayNanos, TimeUnit.NANOSECONDS);
                        }
                    }
                }
                setResultFuture(context);
                if (taskCallback != null) {
                    taskCallback.onComplete(context);
                }
            }

        }

        private void setResultFuture(ScheduleContext<T> context) {
            if (task instanceof CallableTask) {
                if (context.getThrowable() != null) {
                    resultFuture.completeExceptionally(context.getThrowable());
                } else {
                    @SuppressWarnings("unchecked")
                    T result = (T) context.getResult();
                    resultFuture.complete(result);
                }
            } else {
                // RunnableTask 也视为完成（无结果）
                resultFuture.complete(null);
            }
        }

        /**
         * 任务触发后状态的设置
         *
         * @param isCompleted
         * @param oldStatus
         * @param shouldReschedule
         * @return
         */
        private boolean changeStatusAfterExecute(boolean isCompleted, TaskStatus oldStatus, boolean shouldReschedule) {
            if (isCompleted || oldStatus == TaskStatus.CANCELLED) {
                TaskStatus finalStatus = isCompleted ? TaskStatus.COMPLETED : TaskStatus.CANCELLED;
                statusOrdinal.set(finalStatus.ordinal());
            } else if (oldStatus == TaskStatus.CLOSED) {
                statusOrdinal.set(TaskStatus.COMPLETED.ordinal());
            } else if (type == TaskType.ONE_TIME) {
                statusOrdinal.set(TaskStatus.COMPLETED.ordinal());
            } else {
                statusOrdinal.set(TaskStatus.SCHEDULED.ordinal());
                shouldReschedule = true;
            }
            if (!shouldReschedule) {
                cleanup(taskId);
            }
            return shouldReschedule;
        }

        /**
         *
         * @return
         */
        private long calculateNextDelayNanos() {
            ZonedDateTime now = ZonedDateTime.now();
            switch (type) {
                case FIXED_INTERVAL:
                    return intervalNanos;
                case FIBONACCI_INTERVAL:
                    long nextFibNanos = fibPrevNanos + fibCurrNanos;
                    fibPrevNanos = fibCurrNanos;
                    fibCurrNanos = nextFibNanos;
                    return nextFibNanos;
                case BACK_OFF:
                    // 计算下一次执行时间 , 指数退避算法
                    // 计算公式：interval * (2 ^ executionCount)
                    return intervalNanos * (1 << executionCount.get());
                case CRON:
                    if (cronExecutionTime == null) {
                        return -1;
                    }
                    // 计算从现在开始到下一次执行的时间
                    Optional<ZonedDateTime> nextExecution = cronExecutionTime.nextExecution(now);
                    if (nextExecution.isPresent()) {
                        long delayMillis = Duration.between(now, nextExecution.get()).toMillis();
                        // System.out.println("Task " + taskId + " next execution in " + delayMillis + " ms");
                        return TimeUnit.MILLISECONDS.toNanos(delayMillis);
                    } else {
                        // 如果无法计算下一次执行时间（例如，Cron 表达式无效或已过期）
                        log.warn("Could not calculate next execution time for cron task " + taskId);
                        return -1; // 不再调度
                    }
                case ONE_TIME:
                default:
                    return -1; // ONE_TIME 任务不应重新调度
            }
        }

        public TaskStatus getStatus() {
            return TaskStatus.fromOrdinal(statusOrdinal.get());
        }

        public int getExecutionCount() {
            return executionCount.get();
        }

        /**
         * 清理任务资源：从注册表和键映射中移除
         */

    }

    private void cleanup(String taskId) {
        TaskWrapper taskWrapper = taskRegistry.remove(taskId);
        if (taskWrapper != null && taskWrapper.timeout != null) {
            taskWrapper.timeout.cancel();
        }

    }

    // --- 公共调度 API ---
    private String generateTaskId(String taskKey) {
        if (taskKey == null || taskKey.isEmpty()) {
            return "scheduler_task_" + UUID.randomUUID().toString().replace("-", "");
        }
        return "scheduler_task_" + taskKey + "_" + UUID.randomUUID().toString().replace("-", "");
    }

    /**
     * 通用调度方法 (带防重键)
     *
     * @param taskKey 任务的唯一标识键，用于防止重复添加。如果为null或空，则不进行防重检查。
     * @return 任务ID，如果调度失败（队列满或任务重复）则返回null。
     */
    private <T> String schedule(String taskKey, Task<T> task,
                                TaskType type, long delay, TimeUnit delayUnit,
                                long interval, TimeUnit intervalUnit,
                                String cronExpression, int maxExecutions
            , TaskCallback taskCallback) {

        // 1. 检查队列限制
        if (maxPendingTasks > 0 && taskRegistry.size() >= maxPendingTasks) {
            throw new RuntimeException("Task scheduling rejected: Max pending tasks limit (" + maxPendingTasks + ") reached.");
        }


        // 3. 验证参数
        long intervalNanos = checkParams(type, interval, intervalUnit, cronExpression);

        // 4. 创建任务并注册
        String taskId = generateTaskId(taskKey);
        TaskWrapper taskWrapper = new TaskWrapper(taskId, taskKey, task, type, intervalNanos, cronExpression, maxExecutions, taskCallback);

        taskRegistry.put(taskId, taskWrapper);

        // 5. 首次调度
        Long initialDelayNanos = getInitialDelayNanos(taskKey, type, delay, delayUnit, cronExpression, taskWrapper, taskId);
        Timeout timeout = hashedWheelTimer.newTimeout(taskWrapper, initialDelayNanos, TimeUnit.NANOSECONDS);
        taskWrapper.timeout = timeout;

        return taskId;
    }

    private long checkParams(TaskType type, long interval, TimeUnit intervalUnit, String cronExpression) {
        long intervalNanos = 0;
        if (type == TaskType.FIXED_INTERVAL
                || type == TaskType.FIBONACCI_INTERVAL
                || type == TaskType.BACK_OFF) {
            if (interval <= 0) {
                throw new IllegalArgumentException("Interval must be positive for " + type);
            }
            intervalNanos = intervalUnit.toNanos(interval);
        }
        if (type == TaskType.CRON) {
            if (StringUtils.isBlank(cronExpression)) {
                throw new IllegalArgumentException("Cron expression must be provided for CRON task type.");
            }
            try {
                // 预先验证 Cron 表达式
                CRON_PARSER.parse(cronExpression);
            } catch (IllegalArgumentException e) {
                throw new IllegalArgumentException("Invalid cron expression: " + cronExpression, e);
            }
        }
        return intervalNanos;
    }

    private Long getInitialDelayNanos(String taskKey, TaskType type, long delay, TimeUnit delayUnit, String cronExpression, TaskWrapper taskWrapper, String taskId) {
        long initialDelayNanos = delayUnit.toNanos(delay);
        if (type != TaskType.CRON) {
            return initialDelayNanos;
        }
        // 对于 CRON 任务，忽略传入的 delay/delayUnit
        // 直接计算到下一次执行点的延迟作为初始延迟
        long cronInitialDelayNanos = taskWrapper.calculateNextDelayNanos();
        if (cronInitialDelayNanos < 0) {
            // 如果 Cron 表达式没有未来时间点
            // 清理已注册但无法调度的任务
            cleanup(taskId);
            throw new RuntimeException("Invalid cron expression,could not be scheduled as it has no future execution times: " + cronExpression);
        }
        initialDelayNanos = cronInitialDelayNanos;
        return initialDelayNanos;
    }


    @Override
    public <T> String schedule(TaskConfig<T> taskConfig) {
        return schedule(taskConfig.getTaskKey(),
                taskConfig.getTask(), taskConfig.getType(), taskConfig.getInitialDelay()
                , taskConfig.getDelayUnit(), taskConfig.getInterval(), taskConfig.getIntervalUnit(),
                taskConfig.getCronExpression(), taskConfig.getMaxExecutions(), taskConfig.getCallback()
        );
    }


    // --- 任务控制 API ---
    @Override
    public boolean cancel(String taskId) {
        TaskWrapper taskWrapper = taskRegistry.get(taskId);
        if (taskWrapper == null) {
            return false;
        }
        long oldOrdinal = taskWrapper.statusOrdinal.getAndSet(TaskStatus.CANCELLED.ordinal());
        TaskStatus oldStatus = TaskStatus.fromOrdinal(oldOrdinal);
        if (oldStatus != TaskStatus.CANCELLED && oldStatus != TaskStatus.COMPLETED) {
            cleanup(taskId);
            return true;
        }
        return false;
    }


    @Override
    public TaskStatus getTaskStatus(String taskId) {
        TaskWrapper taskWrapper = taskRegistry.get(taskId);
        return taskWrapper != null ? taskWrapper.getStatus() : null;
    }

    @Override
    public int getTaskExecutionCount(String taskId) {
        TaskWrapper taskWrapper = taskRegistry.get(taskId);
        return taskWrapper != null ? taskWrapper.getExecutionCount() : -1;
    }

    // --- 辅助方法 ---

    /**
     * 停止调度器，停止时间轮并清理所有任务。
     * 注意：这是一个强力停止，可能会中断正在运行的任务。
     */
    @Override
    public void shutdown() {
        hashedWheelTimer.stop();
        taskRegistry.clear();
    }


    @Override
    public long getPendingTaskCount() {
        return hashedWheelTimer.pendingTimeouts();
    }

}
